package com.flink.top_N;

import com.flink.datasource.UserSource;
import com.flink.entity.UrlViewCount;
import com.flink.entity.User;
import com.flink.window.WindowFunctionDemo;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;

/**
 * 描述:
 * TODO 使用定时器并用窗口结束时间分组统计，这部分非常重要
 *
 * @author yanzhengwu
 * @create 2022-08-14 17:51
 */
public class TopNExample {
    public static void main(String[] args) throws Exception {

        //声明执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //测试为了保证元素顺序并行度设置为1，可以理解为单线程执行
        env.setParallelism(1);
        //设置水位线生成的间隔 这里给的是100毫秒 ,flink 默认是200毫秒 ，flink 可以达到毫秒级别的效率
        //env.getConfig().setAutoWatermarkInterval(100);


        //TODO 无序流的watermark生成策略
        DataStream<User> stream = env.addSource(new UserSource(100))       //生成水位线和时间戳的策略对象
                .assignTimestampsAndWatermarks(
                        //返回一个具体的策略对象(TODO 这里是乱序流的处理方法，给了一个延迟2秒的策略，也可由理解为 数据延迟多长时间能够全部到位)
                        WatermarkStrategy.<User>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                                //返回策略对象的具体实现
                                .withTimestampAssigner(new SerializableTimestampAssigner<User>() {
                                    /**
                                     * 此方法为指定以事件时间的具体时间戳字段
                                     *
                                     * @param element
                                     * @param recordTimestamp
                                     * @return 返回的则是一个毫秒数的时间戳
                                     */
                                    @Override
                                    public long extractTimestamp(User element, long recordTimestamp) {
                                        return element.getTimestamp();
                                    }
                                }));
        //先通过url进行分组统计窗口中的访问次数
        SingleOutputStreamOperator<UrlViewCount> viewCount = stream.keyBy(User::getProd)
                .window(TumblingEventTimeWindows.of(Time.seconds(5), Time.seconds(1)))
                .aggregate(new UrlCountDemo.UrlViewCountAgg(), new UrlCountDemo.UrlViewCountResult());
        viewCount.print("url count   ");
        //再通过窗口结束时间再次分组排序统计前几名的访问出次数，并在其中使用定时器通过watermark水位线触发输出结果
        viewCount.keyBy(UrlViewCount::getWindowEnd)
                .process(new TopNProcessResult(2))
                .print();


        env.execute();


    }

    /**
     * 由于windowAll会造成数据倾斜不能分布式执行计算所以 为了使用分布式进行处理和汇总数据
     * 使用处理函数和状态列表来汇总和排序所有数据，这样可以使用并行的处理方式来执行
     */
    public static class TopNProcessResult extends KeyedProcessFunction<Long, UrlViewCount, String> {
        //定义取前几名数据
        private Integer n;

        public TopNProcessResult(Integer n) {
            this.n = n;
        }

        //声明状态列表,此状态为运行时状态，只在运行时进行获取一次
        ListState<UrlViewCount> listState;


        @Override
        public void processElement(UrlViewCount value, Context ctx, Collector<String> out) throws Exception {
            //将数据保存至状态列表
            listState.add(value);
            //注册定时器，目的是数据全部到齐后直接执行计算，窗口结束时间+1 ms的时间则为当前窗口数据已经到齐
            ctx.timerService().registerEventTimeTimer(ctx.getCurrentKey() + 1);

        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            List<UrlViewCount> resultList = new ArrayList<>();
            //将状态中的数据进行遍历处理后进行排序
            listState.get().forEach(resultList::add);

            resultList.sort(new Comparator<UrlViewCount>() {
                @Override
                public int compare(UrlViewCount o1, UrlViewCount o2) {
                    //降序排列==后者 - 前者 （此处数量过大会造成数据不准确，实际生产环境下要考量）
                    return o2.getCount().intValue() - o1.getCount().intValue();
                }
            });

            StringBuilder sb = new StringBuilder();

            sb.append("----------------------\n")
                    //.append("窗口开始时间：" + new Timestamp(ctx.getCurrentKey()))
                    .append("窗口结束时间：" + new Timestamp(ctx.getCurrentKey()) + "\n");
            //TODO 这里遍历时 使用我定义的n 进行获取前几名的的排名
            for (int i = 0; i < n; i++) {
                if (i < resultList.size()) {
                    UrlViewCount t = resultList.get(i);
                    String str = "No." + (i + 1) + "\n"
                            + "count:" + t.getCount() + "\n"
                            + "end:" + t.getWindowEnd() + "\n"
                            + "url:" + t.getUrl() + "\n";
                    sb.append(str);
                }


            }
            sb.append("----------------------\n");
            out.collect(sb.toString());
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            //在运行时环境中获取状态列表
            listState = getRuntimeContext().getListState(
                    //此为描述器，声明状态列表名称和类型
                    new ListStateDescriptor<UrlViewCount>("url-count-list", Types.POJO(UrlViewCount.class))
            );
        }
    }
}
